package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DorisSinkFunction;
import com.atguigu.edu.realtime.beans.NewUserBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

public class DwsUserEnroll {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        env.enableCheckpointing(5000L);

        String topic = "dwd_user_enroll";
        String groupId = "DwsUserEnroll";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaStrDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaStrDs.map(JSON::parseObject);

        SingleOutputStreamOperator<JSONObject> fullTsDs = jsonObjDs.map(
                new MapFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
                        String strTime = value.getString("create_time");
                        Timestamp timestamp = Timestamp.valueOf(strTime);
                        long time = timestamp.getTime();
                        value.put("create_time", time);
                        return value;
                    }
                }
        );
        SingleOutputStreamOperator<JSONObject> withWatermarkDs = fullTsDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("create_time");
                                    }
                                }
                        )
        );
        SingleOutputStreamOperator<JSONObject> processDs = withWatermarkDs.process(
                new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        LocalDate now = LocalDate.now();
                        String format = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
                        Long createTime = value.getLong("create_time");
                        String date = DateFormatUtil.toDate(createTime);
                        if (format.equals(date)) {
                            out.collect(value);
                        }
                    }
                }
        );


        AllWindowedStream<JSONObject, TimeWindow> windowDs = processDs.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(3)));

        SingleOutputStreamOperator<NewUserBean> aggregateDs = windowDs.aggregate(
                new AggregateFunction<JSONObject, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(JSONObject value, Long accumulator) {
                        return ++accumulator;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                },
                new AllWindowFunction<Long, NewUserBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<Long> values, Collector<NewUserBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toDate(window.getStart());
                        for (Long value : values) {
                            out.collect(new NewUserBean(
                                    stt,
                                    edt,
                                    curDate,
                                    value
                            ));
                        }
                    }
                }
        );

aggregateDs.addSink(new DorisSinkFunction<NewUserBean>("dws_user_enroll_window"));
 aggregateDs.print(">>>");
           env.execute();
    }
}
